package com.intct.flink;

import com.intct.hbase.bean.Student;
import com.intct.hbase.bean.User;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * @author gufg
 * @since 2025-06-27 09:26
 */
public class ConnectDemo {
    public static void main(String[] args) throws Exception {
        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        /**
         * Connect
         *  1 只能两流连接
         *  2 连接流之后,还需要自己手动处理.调用 map flatMap process处理
         *  3 数据流类型可以不一致
         */

        // 数据源 1
        DataStreamSource<Student> stuDS1 = env.fromElements(
                new Student("c01", "s01", 88L),
                new Student("c01", "s02", 77L),
                new Student("c01", "s03", 66L)
        );

        // 数据源 2
        DataStreamSource<User> stuDS2 = env.fromElements(
                new User(1, "s01", "01"),
                new User(2, "s02", "02"),
                new User(3, "s05", "03")
        );

        // Connect操作
        ConnectedStreams<Student, User> connect = stuDS1.connect(stuDS2);

        SingleOutputStreamOperator<String> mapDS = connect.map(
                /**
                 * 三个参数:
                 *  第一参数: 第一个流的数据
                 *  第二参数: 第二个流的数据
                 *  第三参数: 返回值
                 */
                new CoMapFunction<Student, User, String>() {
                    @Override
                    public String map1(Student value) throws Exception {
                        return value.getClassId();
                    }

                    @Override
                    public String map2(User value) throws Exception {
                        return value.getName();
                    }
                });

        // 输出
        mapDS.print();

        // 启动作业
        env.execute();
    }
}
